package com.aries.dubbo.like.rpc.server.bootstrap;

import com.aries.dubbo.like.rpc.common.codec.ByteBufConst;
import com.aries.dubbo.like.rpc.common.util.ClassUtil;
import com.aries.dubbo.like.rpc.server.handler.RpcServerHandler;
import com.aries.dubbo.like.rpc.service.util.discover.zk.ZookeeperInstance;
import com.aries.dubbo.like.rpc.service.util.discover.zk.ZookeeperRegistry;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;

/**
 * Created with IntelliJ IDEA.
 * Author: aries
 * Date: 2018/8/23
 * Description:
 */
@Slf4j
public class RpcServerBootStrap {

    private static ChannelFuture channelFuture;

    private static EventLoopGroup bossG;
    private static EventLoopGroup workerG;
    private static Thread threadG;

    public void start(int port, String zkAddress, String basePath, Thread thread) {
        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup worker = new NioEventLoopGroup();
        bossG = boss;
        workerG = worker;
        //开启该server的线程，在com.aries.rpc.spring.starter.service.helper.RpcServerConfiguration中定义的。
        threadG = thread;
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(boss, worker)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            /**
                             * tcp粘包
                             */
                            socketChannel
                                    .pipeline()
                                    .addLast(new DelimiterBasedFrameDecoder(10000, Unpooled.copiedBuffer(ByteBufConst.Delimiter())));
                            socketChannel.pipeline().addLast(new RpcServerHandler());
                        }
                    });
            //启动服务端
            ChannelFuture future = bootstrap.bind(port).sync();
            log.info("rpc server has started on port:" + port);
            //设置zookeeper的地址和端口
            ZookeeperInstance.setZkClient(zkAddress);
            ClassUtil.setBasePath(basePath);
            //注册本地服务
            ZookeeperRegistry.register(port);
            channelFuture = future.channel().closeFuture();
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            log.error("error to start rpc server ", e);
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }

    /**
     * 关闭连接，线程池和启动容器的线程。
     */
    public static void shutDown() {
        channelFuture.channel().close();
        bossG.shutdownGracefully();
        workerG.shutdownGracefully();
        threadG.interrupt();
    }
}
